/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;

public class ColumnLoader<VT extends IValueType> {

  private final CabinetLoader cabinetLoader;
  private final ICardLoader<VT> cardLoader;
  private final IColumnIndexer<VT> columnIndexer;
  private final ColumnSegmentInfo<VT> columnInfo;
  private final String columnKey;
  private final DrawerLoader<VT> drawerLoader;
  private final ByteBuffer headerDrawer;

  private int drawerCount;
  private int entryCount;

  public ColumnLoader(final CabinetLoader cabinetLoader,
                      final ICardLoader<VT> cardLoader,
                      final DrawerLoader<VT> drawerLoader,
                      final IColumnIndexer<VT> columnIndexer,
                      final ColumnSegmentInfo<VT> columnInfo, final String columnKey) {
    this.cabinetLoader = cabinetLoader;
    assert this.cabinetLoader != null;

    this.cardLoader = cardLoader;
    assert this.cardLoader != null;

    this.drawerLoader = drawerLoader;
    assert this.drawerLoader != null;

    this.columnIndexer = columnIndexer;
    assert this.columnIndexer != null;

    this.columnInfo = columnInfo;
    assert this.columnInfo != null;

    this.columnKey = columnKey;
    assert this.columnKey != null;

    this.drawerCount = 0;
    this.headerDrawer = ByteBuffer.allocate(ColumnStore.BYTES_IN_DRAWER);
  }

  public void appendBufferedValue(final IValueReference<VT> value) {
    if (!this.cardLoader.appendBufferedValue(value)) {
      if (this.cardLoader.fileCard(this.drawerLoader)) {
        this.cardLoader.reset(this.entryCount);
      } else {
        if (this.drawerLoader.fileDrawer(this)) {
          this.drawerLoader.reset(this.entryCount);
        } else {
          String message = "Failed to flush drawer into buffer file!";
          throw new RuntimeException(message);
        }
        if (this.cardLoader.fileCard(this.drawerLoader)) {
          this.cardLoader.reset(this.entryCount);
        } else {
          String message = "Failed to append to drawer following drawer flush!";
          throw new RuntimeException(message);
        }
      }
      if (!this.cardLoader.appendBufferedValue(value)) {
        String message = "Failed to append to card following card flush!";
        throw new RuntimeException(message);
      }
    }
    this.entryCount++;
  }

  public boolean appendDrawer(final byte[] array,
                              final ColumnSegmentInfo<VT> drawerInfo,
                              final int dataCardCount, final int indexCardCount) {
    // TODO (nbales) - Combine drawers that are not full, but not yet.
    assert indexCardCount >= 1;
    int drawerAddress = this.cabinetLoader.writeDrawer(array);

    int drawerHeaderAddress = dataCardCount + indexCardCount - 1;
    assert drawerAddress < 16777216;
    assert drawerHeaderAddress < 256;
    drawerAddress = (drawerAddress << 8) | drawerHeaderAddress;

    this.drawerCount++;
    this.columnInfo.append(drawerInfo);

    this.columnIndexer.indexDrawer(drawerInfo, drawerAddress);
    return true;
  }

  public void close() {
    // nothing to do
  }

  public int fileColumn() {
    // File any unfinished cards and drawers.
    this.cardLoader.fileCard(this.drawerLoader);
    this.drawerLoader.fileDrawer(this);

    int cardsRequired = this.columnIndexer.cardsRequired();

    if (cardsRequired > ColumnStore.CARDS_IN_DRAWER) {
      String message = "Column header/index exceeds one drawer.";
      throw new RuntimeException(message);
    }

    // TODO (nbales) - Currently each column index occupies its own page. It
    // would be better if column indexes, particularly short ones, were lumped
    // onto other pages (e.g. on empty page from end of column or with other
    // short column indices).
    int indexStartCard = 0;
    int cardsConsumed =
        this.columnIndexer.writeIndex(this.headerDrawer, indexStartCard);
    int headerCardAddress = indexStartCard + cardsConsumed - 1;
    int indexDrawerAddress =
        this.cabinetLoader.writeDrawer(this.headerDrawer.array());

    // Make sure the address values will fit in the 32bit encoding
    assert headerCardAddress < 256;
    assert indexDrawerAddress < 16777216;

    int columnAddress = headerCardAddress | (indexDrawerAddress << 8);

    return columnAddress;
  }

  public String getKey() {
    return this.columnKey;
  }

  public void open() {
    this.drawerLoader.initialize();
    this.cardLoader.initialize();
  }

}
